package com.example.rocketmqspringbootdemo.service.producer.impl;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

//自2.1.0版开始，一个Project中的一个RocketMQTemplate只能有一个TransactionMQProducer来发送所有类型的事务消息
@Service
@RocketMQTransactionListener
class TransactionListenerImpl implements RocketMQLocalTransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<String, Integer>();

    //事务执行逻辑
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        System.out.printf("事务程序开始执行......, tansactionId=%s %n", transId);
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(transId, status);
        if (status == 0) {
            System.out.printf("模拟事务执行成功！ \n");
            return RocketMQLocalTransactionState.COMMIT;
        }
        if (status == 1) {
            System.out.printf("模拟事务执行失败! \n");
            return RocketMQLocalTransactionState.ROLLBACK;
        }

        System.out.printf("模拟事务执行状态未知! \n");
        return RocketMQLocalTransactionState.UNKNOWN;
    }

    //回查逻辑
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        RocketMQLocalTransactionState retState = RocketMQLocalTransactionState.COMMIT;
        Integer status = localTrans.get(transId);
        if (null != status) {
            switch (status) {
                case 0:
                    retState = RocketMQLocalTransactionState.COMMIT;
                    break;
                case 1:
                    retState = RocketMQLocalTransactionState.ROLLBACK;
                    break;
                case 2:
                    retState = RocketMQLocalTransactionState.COMMIT;
                    break;
            }
        }
        System.out.printf("检查程序执行......" + " msgTransactionId=%s, TransactionState=%s status=%s %n", transId, retState, status);
        return retState;
    }
}
